package cn.doitedu.rtdw.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class JOB05_累计销售额大屏指标 {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        /* 开启flinkSql底层自带的提前窗口触发器，每1分钟触发一次*/
        TableConfig config = tenv.getConfig();
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled", true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay", "60000 ms");


        // 建表，映射业务库中的“订单主表”,用cdc连接器
        tenv.executeSql(
                "CREATE TABLE order_mysql (    " +
                        " id            INT ,             " +
                        " status        INT ,             " +
                        " total_amount  decimal(10,0) ,   " +
                        " pay_amount    decimal(10,0) ,   " +
                        " create_time   timestamp(3)  ,   " +
                        " payment_time  timestamp(3)  ,   " +
                        " delivery_time timestamp(3)  ,   " +
                        " confirm_time  timestamp(3)  ,   " +
                        " modify_time   timestamp(3)  ,   " +
                        " rt as modify_time ,             " +
                        " watermark for rt as rt - interval '0' second ,   " +
                        " PRIMARY KEY (id) NOT ENFORCED            " +
                        " ) WITH (                                 " +
                        "     'connector' = 'mysql-cdc',               " +
                        "     'hostname' = 'doitedu'   ,               " +
                        "     'port' = '3306'          ,               " +
                        "     'username' = 'root'      ,               " +
                        "     'password' = 'root'      ,               " +
                        "     'database-name' = 'realtimedw' ,         " +
                        "     'table-name' = 'oms_order'               " +
                        ")"
        );


        tenv.executeSql(
                "SELECT                                               " +
                "    TUMBLE_START(rt,INTERVAL '24' HOUR) as window_start,      " +
                "    TUMBLE_END(rt,INTERVAL '24' HOUR) as window_end,          " +
                "    sum(total_amount) as total_amount,                        " +
                "    sum(pay_amount) as total_pay_amount,                      " +
                "    sum(total_amount - pay_amount) as total_discount_amount   " +
                "FROM order_mysql                                              " +
                "GROUP BY                                                      " +
                "    TUMBLE(rt,INTERVAL '24' HOUR)").print();

    }
}
